package com.boarsoft.boar.flow.monitor;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

import org.bson.Document;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import com.boarsoft.common.Util;
import com.boarsoft.flow.trace.FlowStep;
import com.boarsoft.flow.trace.FlowTrace;
import com.boarsoft.message.bean.Message;
import com.boarsoft.message.core.MessageService;
import com.boarsoft.mongodb.core.MongoDao;
import com.mongodb.client.MongoCollection;

/**
 * 监控消息接收器，对请求端发送对消息进行处理和存储
 * 
 * @author Mac_J
 */
public class FlowMonitorImpl implements MessageService {
	private static final Logger log = LoggerFactory.getLogger(FlowMonitorImpl.class);

	@Autowired
	protected MongoDao mongoDao;
	
	/** 将消息的key映射到MongoDB的db.collection */
	protected Map<String, String> mapping = new HashMap<String, String>();

	/** 缓存 MongoCollection */
	protected Map<String, MongoCollection<Document>> collMap = new ConcurrentHashMap<String, MongoCollection<Document>>();

	@Override
	public void put(Message m) throws Exception {
		log.info("Received message: {}", m);
		// key = catalog.flowId
		MongoCollection<Document> coll = this.getCollection(m.getKey());
		if (coll == null) {
			log.warn("Can not find mongodb collection for target queue {}", m.getTarget());
			return;
		}
		// Document doc = this.document((FlowTrace) msg.getContent());
		Document doc = Document.parse((String) m.getContent());
		doc.put("_id", m.getId());
		doc.put("env", FlowMonitorConfig.APP_ENV);
		doc.put("addr", m.getFromAddr());
		coll.insertOne(doc);
	}

	protected MongoCollection<Document> getCollection(String key) {
		MongoCollection<Document> mc = collMap.get(key);
		if (mc == null) {
			synchronized (collMap) {
				mc = collMap.get(key);
				if (mc == null) {
					String n = mapping.get(key);
					if (Util.strIsEmpty(n)) {
						n = key;
					}
					String[] a = n.split("\\.");
					if (a.length > 1) {
						mc = mongoDao.getCollection(a[0], a[1]);
					}
					collMap.put(key, mc);
				}
			}
		}
		return mc;
	}

	protected Document document(FlowTrace trace) {
		Document doc = new Document();
		doc.put("flowId", trace.getFlowId());
		doc.put("id", trace.getFlowId());
		doc.put("threadName", trace.getThreadName());
		List<Document> stepLt = new LinkedList<Document>();
		for (FlowStep s : trace.getSteps()) {
			Document d = new Document();
			d.put("code", s.getCode());
			d.put("threadName", s.getThreadName());
			d.put("startTime", s.getStartTime());
			d.put("endTime", s.getEndTime());
			List<Document> z = new LinkedList<Document>();
			for (FlowTrace t : s.getTraces()) {
				z.add(this.document(t));
			}
			d.put("traces", z);
			d.put("traces", s.getThreadName());
			d.put("input", s.getInput()); // byte[]
			d.put("output", s.getOutput()); // byte[]
			stepLt.add(d);
		}
		doc.put("steps", stepLt);
		return doc;
	}

	public Map<String, String> getMapping() {
		return mapping;
	}

	public void setMapping(Map<String, String> mapping) {
		this.mapping = mapping;
	}
}
